package club.yuit.rmq.service;

import club.yuit.rmq.config.RMQConfigure;
import club.yuit.rmq.model.BrokerDetails;
import club.yuit.rmq.model.CUTConfig;
import club.yuit.rmq.model.TopicDetails;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.shaded.com.google.common.base.Throwables;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.producer.DefaultMQProducer;

import java.lang.reflect.InvocationTargetException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/**
 * @author: yuit
 * @date: 2023/12/04 16:33
 */
@Slf4j
public class RMQClusterService {

    private MQAdminExt admin;
    private RMQConfigure configure;


    public RMQClusterService(RMQConfigure configure, MQAdminExt admin) {
        this.admin = admin;
        this.configure = configure;
    }

    public Set<String> getClusterList() throws Exception {
        ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
        return clusterInfo.getClusterAddrTable().keySet();
    }


    public Map<String, List<BrokerDetails>> brokerList() throws Exception {
        ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
        Map<String, BrokerData> brokerAddrTable = clusterInfo.getBrokerAddrTable();

        Map<String, List<BrokerDetails>> brokerDetails = Maps.newHashMap();


        for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {

            List<BrokerDetails> detailsList = getDetailsList(brokerDetails, brokerData.getCluster());
            BrokerDetails bks = new BrokerDetails();
            bks.setCluster(brokerData.getCluster());
            bks.setName(brokerData.getBrokerName());
            for (Map.Entry<Long/* brokerId */, String/* broker address */> brokerAddr : brokerData.getBrokerAddrs().entrySet()) {


                KVTable kvTable = admin.fetchBrokerRuntimeStats(brokerAddr.getValue());
                HashMap<String, String> table = kvTable.getTable();
                table.put("brokerAddr", brokerAddr.getValue());
                bks.setDetails(table);
                bks.setVersionDec(table.get("brokerVersionDesc"));
                bks.setId(brokerAddr.getKey());
                bks.setAddr(brokerAddr.getValue());
                detailsList.add(bks);
            }

        }
        return brokerDetails;
    }

    private List<BrokerDetails> getDetailsList(Map<String, List<BrokerDetails>> brokerDetails, String cluster) {
        return brokerDetails.computeIfAbsent(cluster, k -> new ArrayList<>());
    }


    public List<TopicDetails> getTopics() throws Exception {
        TopicList topicList = admin.fetchAllTopicList();
        List<TopicDetails> tps = new ArrayList<>();

        for (String topic : topicList.getTopicList()) {
            TopicDetails details = new TopicDetails();
            details.setName(topic);
            TopicRouteData topicRouteData = admin.examineTopicRouteInfo(topic);
            List<BrokerData> brokerDatas = topicRouteData.getBrokerDatas();
            details.setBrokerData(brokerDatas);
            List<TopicConfig> topicConfigs = new ArrayList<>();
            for (BrokerData data : brokerDatas) {
                details.getCluster().add(data.getCluster());
                TopicConfig topicConfig = admin.examineTopicConfig(data.selectBrokerAddr(), topic);
                topicConfigs.add(topicConfig);
            }
            details.setTopicConfig(topicConfigs);
            tps.add(details);
        }


        // 系统topic
        TopicList systemTopicList = getSystemTopicList();
        Set<String> sysTopics = systemTopicList.getTopicList();

        for (TopicDetails tp : tps) {
            if (sysTopics.contains(tp.getName())) {
                String name = String.format("%s%s", "%SYS%", tp.getName());
                tp.setName(name);
            }
        }

        return tps;
    }


    private TopicList getSystemTopicList() {
        RPCHook rpcHook = null;
        boolean isEnableAcl = !StringUtils.isEmpty(configure.getAccessKey()) && !StringUtils.isEmpty(configure.getSecretKey());
        if (isEnableAcl) {
            rpcHook = new AclClientRPCHook(new SessionCredentials(configure.getAccessKey(), configure.getSecretKey()));
        }
        DefaultMQProducer producer = buildDefaultMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, false);
        producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
        producer.setNamesrvAddr(configure.getNamesrvAddr());

        try {
            producer.start();
            return producer.getDefaultMQProducerImpl().getmQClientFactory().getMQClientAPIImpl().getSystemTopicList(20000L);
        } catch (Exception e) {
            Throwables.throwIfUnchecked(e);
            throw new RuntimeException(e);
        } finally {
            producer.shutdown();
        }
    }

    public DefaultMQProducer buildDefaultMQProducer(String producerGroup, RPCHook rpcHook, boolean traceEnabled) {
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(producerGroup, rpcHook, traceEnabled, TopicValidator.RMQ_SYS_TRACE_TOPIC);
        defaultMQProducer.setUseTLS(configure.isUseTLS());
        return defaultMQProducer;
    }


    public Map<String, String> getBrokerConfig(String brokerAddr) throws Exception {
        Properties p = this.admin.getBrokerConfig(brokerAddr);
        Set<Map.Entry<Object, Object>> entries = p.entrySet();
        Map<String, String> result = new HashMap<>();

        for (Map.Entry<Object, Object> entry : entries) {
            result.put(entry.getKey().toString(), entry.getValue().toString());
        }
        return result;
    }


    public void createOrUpdateTopic(CUTConfig config) throws Exception {


        String cluster = config.getCluster();

        ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
        // master节点地址
        List<String> brokerAddrs = clusterInfo
                .getBrokerAddrTable()
                .values()
                .stream()
                .filter(v -> filterBroker(v, cluster, config.getBrokers()))
                .map(v -> v.getBrokerAddrs().get(0L)).toList();

        log.debug("create or update topic, master address: [{}]", brokerAddrs);

        for (String brokerAddr : brokerAddrs) {
            this.admin.createAndUpdateTopicConfig(brokerAddr, config);
        }
    }

    private boolean filterBroker(BrokerData data, String cluster, List<String> brokers) {
        String masterAddr = data.getBrokerAddrs().get(0L);

        // 不是master 节点 不能操作Topic
        if (masterAddr == null) {
            return false;
        }

        if (cluster != null) {
            if (cluster.equals(data.getCluster())) {
                if (brokers != null && !brokers.isEmpty()) {
                    return brokers.contains(data.getBrokerName());
                }
            }
            return false;

        } else {
            if (brokers != null && !brokers.isEmpty()) {
                return brokers.contains(data.getBrokerName());
            }

            return true;
        }


    }


}
